package com.hexinfo.dmpro.monitor.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.hexinfo.dmpro.common.utils.StringUtils;
import com.hexinfo.dmpro.monitor.config.MonitorCache;
import com.hexinfo.dmpro.monitor.dao.YarnAppDao;
import com.hexinfo.dmpro.monitor.dao.YarnMetricsDao;
import com.hexinfo.dmpro.monitor.dao.YarnSchedulerDao;
import com.hexinfo.dmpro.monitor.domain.ClusterDomain;
import com.hexinfo.dmpro.monitor.model.*;
import com.hexinfo.dmpro.monitor.model.base.MonitorReq;
import com.hexinfo.dmpro.monitor.model.base.MonitorRsp;
import com.hexinfo.dmpro.monitor.model.cm.DetailResourceInfo;
import com.hexinfo.dmpro.monitor.model.serviceparam.CmTargetServiceModel;
import com.hexinfo.dmpro.monitor.service.serviceparam.CmTargetParamService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.text.DecimalFormat;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author peng
 * @date 2021/9/28 19:24
 */
@Service
@Slf4j
public class YarnService {

    private int SUCESS_CODE = 200;

    private String[] YARN_HEALTH = {"STARTED", "ACTIVE", "STANDBY", "NEW", "RUNNING"};
    public static String YARN_HEALTH_STRING = "STARTED,ACTIVE,STANDBY,NEW,RUNNING";
    public static String YARN_WARNING_STRING = "INITED,UNHEALTHY,DECOMMISSIONING,DECOMMISSIONED";
    public static String YARN_DEADLY_STRING = "NOTINITED,INITIALIZING,REBOOTED,SHUTDOWN,LOST";
    public static String YARN_STOP_STRING = "STOPPED";
    private String[] YARN_WARNING = {"INITED", "UNHEALTHY", "DECOMMISSIONING", "DECOMMISSIONED"};
    //
    private String[] YARN_DEADLY = {"LOST", "NOTINITED", "INITIALIZING", "REBOOTED", "SHUTDOWN"};

    @Autowired
    private YarnAppService yarnAppService;
    @Autowired
    private YarnMetricsService yarnMetricsService;
    @Autowired
    private YarnSchedulerService yarnSchedulerService;
    @Autowired
    private YarnAppDao yarnAppDao;
    @Autowired
    private YarnSchedulerDao yarnSchedulerDao;
    @Autowired
    private YarnMetricsDao yarnMetricsDao;
    @Autowired
    private CmYarnAppService cmYarnAppService;
    @Autowired
    private CmTargetParamService cmTargetParamService;
    @Autowired
    private CmYarnService cmYarnService;

    @Value("${warn.cmYarnResourceWarn}")
    private String cmYarnResourceWarn;
    @Value("${warn.cmYarnCpuUse}")
    private String cmYarnCpuUse;
    @Value("${warn.cmYarnMemoryUse}")
    private String cmYarnMemoryUse;
    @Value("${warn.cmYarnRunTime}")
    private String cmYarnRunTime;
    @Value("${warn.cmYarnRunNumber}")
    private String cmYarnRunNumber;

    public YarnMetricsModel getYarnMetrics1(MonitorReq req, boolean flag) {
        YarnMetricsModel yarnMetricsModel = null;
        List<MonitorReq> monitorReqList = MonitorCache.getMonitorReq(req.getCenterId(), req.getCmFlag(), req.getClusterFlag(), MonitorParam.YARN_METRICS);
        for (MonitorReq monitorReq : monitorReqList) {
            MonitorRsp rsp = yarnMetricsService.execute(monitorReq);
            if (SUCESS_CODE == rsp.getCode()) {
                yarnMetricsModel = (YarnMetricsModel) rsp.getObj();
            }
            if (flag && SUCESS_CODE == rsp.getCode()) {
                req.setCenterName(monitorReq.getCenterName());
                req.setClusterName(monitorReq.getClusterName());
                yarnMetricsModel.setTaskTime(req.getTaskTime());
                yarnMetricsModel.setCenterId(req.getCenterId());
                yarnMetricsModel.setCenterName(req.getCenterName());
                yarnMetricsModel.setCmFlag(req.getCmFlag());
                yarnMetricsModel.setCmName(req.getCmName());
                yarnMetricsModel.setClusterFlag(req.getClusterFlag());
                yarnMetricsModel.setClusterName(req.getClusterName());
                yarnMetricsDao.insert(yarnMetricsModel);

                //数据同时存入最新时刻表   处理查询效率慢问题
                yarnMetricsDao.deleteNow(yarnMetricsModel);
                yarnMetricsDao.insertNow(yarnMetricsModel);
            }
        }
        return yarnMetricsModel;
    }

    /**
     * 运行作业信息
     *
     * @return
     */
    public List<YarnAppModel> getYarnApp1(MonitorReq req, boolean isFlag) {
        //yarn有些数据只能通过手动维护的url表中的http://url:8088/ws/v1/cluster/apps来获取
        List<MonitorReq> monitorReqList = MonitorCache.getMonitorReq(req.getCenterId(), req.getCmFlag(), req.getClusterFlag(), MonitorParam.YARN_APP);
        MonitorRsp rsp = null;
        List<YarnAppModel> yarnAppModelList = null;
        List<YarnAppModel> finishYarnApps = null;
        MonitorReq succMonitorReq = new MonitorReq();
        for (MonitorReq monitorReq : monitorReqList) {
            monitorReq.setStartTime(req.getStartTime());
            monitorReq.setEndTime(req.getEndTime());
            monitorReq.setUser(req.getUser());
            monitorReq.setJobType(req.getJobType());
            monitorReq.setState(req.getState());
            String url = monitorReq.getUrl();
            req.setCenterName(monitorReq.getCenterName());
            req.setClusterName(monitorReq.getClusterName());
            rsp = yarnAppService.execute(monitorReq);
            if (SUCESS_CODE == rsp.getCode()) {
                yarnAppModelList = (List<YarnAppModel>) rsp.getObj();
                //添加完成时间处理逻辑
                monitorReq.setFinishedTimeBegin(req.getStartTime());
                monitorReq.setUrl(url);
                monitorReq.setStartTime(null);
                MonitorRsp finshRsp = yarnAppService.execute(monitorReq);
                if (SUCESS_CODE == finshRsp.getCode()) {
                    finishYarnApps = (List<YarnAppModel>) finshRsp.getObj();
                }
                succMonitorReq = monitorReq;
                succMonitorReq.setUrl(url);
                break;
            }
        }

        Map<String, YarnAppModel> ysApp = new HashMap<>();
        if (yarnAppModelList != null && yarnAppModelList.size() > 0) {
            for (int i = 0; i < yarnAppModelList.size(); i++) {
                ysApp.put(yarnAppModelList.get(i).getId(), yarnAppModelList.get(i));
            }
            if (finishYarnApps != null && finishYarnApps.size() > 0) {
                for (int i = 0; i < finishYarnApps.size(); i++) {
                    ysApp.put(finishYarnApps.get(i).getId(), finishYarnApps.get(i));
                }
            }
        }
        String startDate = LocalDateTime.now().plusHours(1).minusMinutes(6).toString().substring(0, 19).replace("T", " ");
        req.setStartTime(startDate);

        String serviceName = MonitorCache.getServiceName(req.getCenterId(), req.getCmFlag(), req.getClusterFlag(), "yarn");
        //集群没有yarn服务时，直接退出
        if (StrUtil.isBlank(serviceName)){
            return yarnAppModelList;
        }
        req.setServiceName(serviceName);
        List<YarnAppModel> yarnApps = cmYarnAppService.getYarnApps1(req);
        String yarnUrl = succMonitorReq.getUrl();
        if (yarnApps != null && yarnApps.size() > 0) {
            for (int i = 0; i < yarnApps.size(); i++) {
                YarnAppModel yarnAppModel = yarnApps.get(i);
                String appid = yarnAppModel.getId();
                if (ysApp.get(appid) == null) {
                    succMonitorReq.setUrl(yarnUrl);
                    succMonitorReq.setYarnAppId(appid);
                    rsp = yarnAppService.execute(succMonitorReq);
                    YarnAppModel appModel = (YarnAppModel) rsp.getObj();
                    ysApp.put(appid, appModel);
                }
                if (ysApp.get(appid) != null) {
                    yarnAppModel.setApplicationType(ysApp.get(appid).getApplicationType());
                    yarnAppModel.setApplicationDuration(ysApp.get(appid).getApplicationDuration());
                    yarnAppModel.setFinalStatus(ysApp.get(appid).getFinalStatus());
                    yarnAppModel.setState(ysApp.get(appid).getState());
                    yarnAppModel.setAllocatedMB(ysApp.get(appid).getAllocatedMB());
                    yarnAppModel.setAllocatedVCores(ysApp.get(appid).getAllocatedVCores());
                    yarnAppModel.setRunningContainers(ysApp.get(appid).getRunningContainers());
                    yarnAppModel.setQueueUsagePercentage(ysApp.get(appid).getQueueUsagePercentage());
                    yarnAppModel.setClusterUsagePercentage(ysApp.get(appid).getClusterUsagePercentage());
                    yarnAppModel.setMemorySeconds(ysApp.get(appid).getMemorySeconds());
                    yarnAppModel.setVcoreSeconds(ysApp.get(appid).getVcoreSeconds());
                    yarnAppModel.setStartedTime(ysApp.get(appid).getStartedTime());
                    yarnAppModel.setLaunchTime(ysApp.get(appid).getLaunchTime());
                    yarnAppModel.setElapsedTime(ysApp.get(appid).getElapsedTime());
                    yarnAppModel.setClusterId(ysApp.get(appid).getClusterId());
                    yarnAppModel.setTrackingUrl(ysApp.get(appid).getTrackingUrl());
                }
            }
        }

        if (isFlag) {
            if (yarnApps != null && yarnApps.size() > 0) {
                try{
                    //yarn任务资源使用情况监控
                    yarnWarn(yarnApps,req);
                }catch (Exception e){
                    log.error("yarn任务资源使用情况监控异常"+e);
                    e.printStackTrace();
                }
                for (YarnAppModel yarnApp : yarnApps) {
                    try {
                        yarnApp.setCenterId(req.getCenterId());
                        yarnApp.setCenterName(req.getCenterName());
                        yarnApp.setCmFlag(req.getCmFlag());
                        yarnApp.setCmName(req.getCmName());
                        yarnApp.setClusterFlag(req.getClusterFlag());
                        yarnApp.setClusterName(req.getClusterName());
                        yarnApp.setTaskTime(req.getTaskTime());
//                        yarnAppDao.updateInsertApp(yarnApp);
                        yarnAppDao.delInsertApp(yarnApp);
                        yarnAppDao.upInsertApp(yarnApp);
                    }catch (Exception e){
                        log.error("任务监控yarn任务插入数据库报错："+e);
                    }
                }
            }
        }
        return yarnAppModelList;
    }

    /**
     * yarn任务资源使用情况监控
     * @return
     */
    public String yarnWarn(List<YarnAppModel> yarnApps,MonitorReq req) {
        StringBuffer warnAlert = new StringBuffer();
        //yarn任务预警配置信息
        CmTargetServiceModel cpuModel = cmTargetParamService.queryTargetCode(cmYarnCpuUse);
        CmTargetServiceModel memoryModel = cmTargetParamService.queryTargetCode(cmYarnMemoryUse);
        CmTargetServiceModel timeModel = cmTargetParamService.queryTargetCode(cmYarnRunTime);
        CmTargetServiceModel numberModel = cmTargetParamService.queryTargetCode(cmYarnRunNumber);
        //获取集群root队列信息
        List<DetailResourceInfo> DetailResourceInfos = cmYarnService.queryRootPool(req.getCmFlag(), req.getClusterFlag());
        //按照Queue队列转化为Map
        Map<String, DetailResourceInfo> map = DetailResourceInfos.stream()
                .collect(Collectors.toMap(DetailResourceInfo::getQueue, Function.identity()));
        DetailResourceInfo root = map.get("root");
        int runNum = 0;
        for (YarnAppModel yarnApp : yarnApps) {
            String warnLog = "";
            //（1）yarn任务持续时间超过N小时，flink任务不在超时告警内
            if (timeModel != null){
                if (!"Apache Flink".equalsIgnoreCase(yarnApp.getApplicationType())){
                    long num = Long.parseLong(yarnApp.getElapsedTime());
                    long l = Long.parseLong(timeModel.getDefaultValue())*60*1000;
                    if (num > l){
                        double result = (double) num / (1000*60);
                        DecimalFormat decimalFormat = new DecimalFormat("#");
                        warnLog += "持续时间："+decimalFormat.format(result)+"分钟，告警值："+timeModel.getDefaultValue()+"分钟"+"\n";
                    }
                }
            }
            //（2）yarn任务正在执行的任务数超过N个
            if ("RUNNING".equalsIgnoreCase(yarnApp.getState())){
                runNum++;
                DetailResourceInfo queuePool = map.get(yarnApp.getQueue());
                //动态资源池跳过
                if (queuePool != null){
                    //（3）yarn单个任务cpu超过总资源的百分比
                    if (cpuModel != null){
                        double queueCpuPer = ((double) yarnApp.getAllocatedVCores() / Long.parseLong(queuePool.getMaxCores()));
                        double cmCpuPer = ((double) yarnApp.getAllocatedVCores() / Long.parseLong(root.getMaxCores()));
                        //队列资源百分比
                        if (queueCpuPer > Double.parseDouble(cpuModel.getDefaultValue())){
                            warnLog +="CPU使用："+yarnApp.getAllocatedVCores()+"\n";
                            warnLog +="CPU队列使用率："+String.format("%.1f", queueCpuPer*100)+"%"+"\n";
                        }
                        //集群资源百分比
                        if (cmCpuPer > Double.parseDouble(cpuModel.getAlarmValue())){
                            warnLog +="CPU集群使用率："+String.format("%.1f", cmCpuPer*100)+"%"+"\n";
                        }
                    }
                    //（4）yarn单个任务内存超过总资源的百分比
                    if (memoryModel != null){
                        double queueMBPer = ((double) yarnApp.getAllocatedMB() / Long.parseLong(queuePool.getMaxGB()));
                        double cmMBPer = ((double) yarnApp.getAllocatedMB() / Long.parseLong(root.getMaxGB()));
                        //队列资源百分比
                        if (queueMBPer > Double.parseDouble(memoryModel.getDefaultValue())){
                            warnLog +="内存使用："+yarnApp.getAllocatedMB()+"MB"+"\n";
                            warnLog +="内存队列使用率："+String.format("%.1f", queueMBPer*100)+"%"+"\n";
                        }
                        //集群资源百分比
                        if (cmMBPer > Double.parseDouble(memoryModel.getAlarmValue())){
                            warnLog +="内存集群使用率："+String.format("%.1f", cmMBPer*100)+"%"+"\n";
                        }
                    }
                }
            }
            if (StrUtil.isNotBlank(warnLog)){
                //sql截取
                String sql = "";
                if (yarnApp.getName().length() <= 100) {
                    sql = yarnApp.getName();
                } else {
                    sql = yarnApp.getName().substring(0, 100) + "...";
                }
                warnLog = "任务名："+sql+"\n"+
                        "任务id："+yarnApp.getId()+"\n"+
                        "任务类型："+yarnApp.getApplicationType()+"\n"+
                        "池："+yarnApp.getQueue()+"\n"+
                        warnLog+"\n";
                warnAlert.append(warnLog);
            }
        }
        if (numberModel != null){
            int warnNum = Integer.parseInt(numberModel.getDefaultValue());
            if(runNum > warnNum){
                warnAlert.insert(0,"yarn进行中的任务数："+runNum+"，告警值："+warnNum+"\n");
            }
        }
        if(StrUtil.isNotEmpty(warnAlert)){
            warnAlert.insert(0,"【yarn任务资源告警】"+"\n"+"【"+req.getCmName()+"--"+req.getClusterName()+"】"+"\n");
            //发送yarn任务告警，钉钉
            cmTargetParamService.warnWxDingDingAndLog(req.getCmFlag(),req.getCmName(),req.getClusterFlag(),
                    req.getClusterName(),"yarn",warnAlert.toString(),cmYarnResourceWarn);
        }
        return warnAlert.toString();
    }

    /**
     * 队列资源信息
     *
     * @return
     */
    public YarnSchedulerModel getYarnScheduler(MonitorReq req, boolean flag) {
        List<MonitorReq> monitorReqList = MonitorCache.getMonitorReq(req.getCenterId(), req.getCmFlag(), req.getClusterFlag(), MonitorParam.YARN_SCHEDULER);
        YarnSchedulerModel yarnSchedulerModel = null;
        for (MonitorReq monitorReq : monitorReqList) {
            req.setCenterName(monitorReq.getCenterName());
            req.setClusterName(monitorReq.getClusterName());
            MonitorRsp rsp = yarnSchedulerService.execute(monitorReq);
            if (SUCESS_CODE == rsp.getCode()) {
                yarnSchedulerModel = (YarnSchedulerModel) rsp.getObj();
                break;
            }
        }
        if (flag && yarnSchedulerModel != null) {
            Map<String, Object> map = new HashMap();
            List<MonitorResourceBaseModel> resourceList = new ArrayList<>();
            dealScheduler(resourceList, yarnSchedulerModel);
            map.put("list", resourceList);
            map.put("req", req);
            yarnSchedulerDao.insertScheduler(map);

            //数据同时存入最新时刻表   处理查询效率慢问题
            yarnSchedulerDao.deleteSchedulerNow(map);
            yarnSchedulerDao.insertSchedulerNow(map);

        }
        return yarnSchedulerModel;
    }

    /**
     * 递归处理队列信息
     *
     * @return
     */
    private void dealScheduler(List<MonitorResourceBaseModel> resourceList, YarnSchedulerModel yarnScheduler) {
        if (yarnScheduler != null) {
            YarnSchedulerModel.QueuesResources maxResources = yarnScheduler.getMaxResources();
            YarnSchedulerModel.QueuesResources usedResources = yarnScheduler.getUsedResources();
            MonitorResourceBaseModel monitorResourceBaseModel = new MonitorResourceBaseModel();
            monitorResourceBaseModel.setName(yarnScheduler.getQueueName());
            monitorResourceBaseModel.setMaxCores(maxResources.getVCores());
            monitorResourceBaseModel.setMaxMB(maxResources.getMemory());
            monitorResourceBaseModel.setUsedCores(usedResources.getVCores());
            monitorResourceBaseModel.setUsedMB(usedResources.getMemory());
            resourceList.add(monitorResourceBaseModel);
            YarnSchedulerModel.ChildQueues childQueues = yarnScheduler.getChildQueues();
            if (childQueues != null && childQueues.getQueue() != null && childQueues.getQueue().size() > 0) {
                List<YarnSchedulerModel> queues = childQueues.getQueue();
                for (YarnSchedulerModel queue : queues) {
                    dealScheduler(resourceList, queue);
                }
            }
        }
    }
}
